package cn.itcast.czxy.BD18.ml

import cn.itcast.czxy.BD18.bean.BaseMode
import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.classification.{DecisionTreeClassificationModel, DecisionTreeClassifier}
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator
import org.apache.spark.ml.feature.{StringIndexer, VectorAssembler}
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.{Column, DataFrame, functions}
import org.apache.spark.storage.StorageLevel

object USGMode1 extends BaseMode {
  override def setAppName: String = "USGMode1"

  override def setLeven4Id: Int = 151

  override def getNewTag(leve5: DataFrame, hbaseDF: DataFrame): DataFrame = {
    import spark.implicits._
    import org.apache.spark.sql.functions._
    val tbl_orders = hbaseDF.persist(StorageLevel.MEMORY_AND_DISK)
    val tbl_goods = spark.read.format("cn.itcast.czxy.BD18.tools.HBaseDataSource")
      .option("zkHosts", "192.168.10.20")
      .option("zkPort", "2181")
      .option("hbaseTable", "tbl_goods")
      .option("family", "detail")
      .option("selectFields", "cOrderSn,productType,ogColor")
      .load().persist(StorageLevel.MEMORY_AND_DISK)
    //    tbl_orders.show()
    //    tbl_goods.show()
    //    +---------+-------------------+
    //    | memberId|            orderSn|
    //    +---------+-------------------+
    //    | 13823431| ts_792756751164275|
    //
    //    +--------------------+-----------+---------+
    //    |            cOrderSn|productType|  ogColor|
    //    +--------------------+-----------+---------+
    //    |jd_14091818005983607|       烤箱|     白色|
    val orders_goods = tbl_orders.join(tbl_goods, tbl_orders("orderSn") === tbl_goods("cOrderSn")).select("memberId", "productType", "ogColor").persist(StorageLevel.MEMORY_AND_DISK)
    //    orders_goods.show()
    //    +---------+-----------+---------+
    //    | memberId|productType|  ogColor|
    //    +---------+-----------+---------+
    //    | 13823535|       其他|     灰色|
    //    | 13823535|   智能电视|     银色|
    //    | 13823535| 燃气热水器|     粉色|

    val label: Column = functions
      .when('ogColor.equalTo("樱花粉")
        .or('ogColor.equalTo("白色"))
        .or('ogColor.equalTo("香槟色"))
        .or('ogColor.equalTo("香槟金"))
        .or('productType.equalTo("料理机"))
        .or('productType.equalTo("挂烫机"))
        .or('productType.equalTo("吸尘器/除螨仪")), 1) //女
      .otherwise(0) //男
      .alias("gender")

    //决策树算法需要的特征数据不能是字符串类型，但是我们的数据是字符串
    //所以我们需要将这里的字符串特征变为数值类型
    //颜色ID应该来源于字典表,这里简化处理

    //这里的编号，最好是数据库中读取
    val color: Column = functions
      .when('ogColor.equalTo("银色"), 1)
      .when('ogColor.equalTo("香槟金色"), 2)
      .when('ogColor.equalTo("黑色"), 3)
      .when('ogColor.equalTo("白色"), 4)
      .when('ogColor.equalTo("梦境极光【卡其金】"), 5)
      .when('ogColor.equalTo("梦境极光【布朗灰】"), 6)
      .when('ogColor.equalTo("粉色"), 7)
      .when('ogColor.equalTo("金属灰"), 8)
      .when('ogColor.equalTo("金色"), 9)
      .when('ogColor.equalTo("乐享金"), 10)
      .when('ogColor.equalTo("布鲁钢"), 11)
      .when('ogColor.equalTo("月光银"), 12)
      .when('ogColor.equalTo("时尚光谱【浅金棕】"), 13)
      .when('ogColor.equalTo("香槟色"), 14)
      .when('ogColor.equalTo("香槟金"), 15)
      .when('ogColor.equalTo("灰色"), 16)
      .when('ogColor.equalTo("樱花粉"), 17)
      .when('ogColor.equalTo("蓝色"), 18)
      .when('ogColor.equalTo("金属银"), 19)
      .when('ogColor.equalTo("玫瑰金"), 20)
      .otherwise(0)
      .alias("color")


    //类型ID应该来源于字典表,这里简化处理
    val productType: Column = functions
      .when('productType.equalTo("4K电视"), 9)
      .when('productType.equalTo("Haier/海尔冰箱"), 10)
      .when('productType.equalTo("Haier/海尔冰箱"), 11)
      .when('productType.equalTo("LED电视"), 12)
      .when('productType.equalTo("Leader/统帅冰箱"), 13)
      .when('productType.equalTo("冰吧"), 14)
      .when('productType.equalTo("冷柜"), 15)
      .when('productType.equalTo("净水机"), 16)
      .when('productType.equalTo("前置过滤器"), 17)
      .when('productType.equalTo("取暖电器"), 18)
      .when('productType.equalTo("吸尘器/除螨仪"), 19)
      .when('productType.equalTo("嵌入式厨电"), 20)
      .when('productType.equalTo("微波炉"), 21)
      .when('productType.equalTo("挂烫机"), 22)
      .when('productType.equalTo("料理机"), 23)
      .when('productType.equalTo("智能电视"), 24)
      .when('productType.equalTo("波轮洗衣机"), 25)
      .when('productType.equalTo("滤芯"), 26)
      .when('productType.equalTo("烟灶套系"), 27)
      .when('productType.equalTo("烤箱"), 28)
      .when('productType.equalTo("燃气灶"), 29)
      .when('productType.equalTo("燃气热水器"), 30)
      .when('productType.equalTo("电水壶/热水瓶"), 31)
      .when('productType.equalTo("电热水器"), 32)
      .when('productType.equalTo("电磁炉"), 33)
      .when('productType.equalTo("电风扇"), 34)
      .when('productType.equalTo("电饭煲"), 35)
      .when('productType.equalTo("破壁机"), 36)
      .when('productType.equalTo("空气净化器"), 37)
      .otherwise(0)
      .alias("productType")
    val orders_goods_int: DataFrame = orders_goods.select('memberId, productType, color, label).persist(StorageLevel.MEMORY_AND_DISK)
    //    orders_goods_int.show()
    //    +---------+-----------+-----+------+
    //    | memberId|productType|color|gender|
    //    +---------+-----------+-----+------+
    //    | 13823535|          0|   16|     0|
    //    | 13823535|         24|    1|     0|
    //    | 13823535|         30|    7|     0|

    val labelInt = new StringIndexer()
      .setInputCol("gender").setOutputCol("label").fit(orders_goods_int)

    val features = new VectorAssembler()
      .setInputCols(Array("productType", "color")).setOutputCol("features")

    val decisionTree = new DecisionTreeClassifier()
      .setFeaturesCol("features").setPredictionCol("prediction")
      .setMaxDepth(5)

    val pipeline = new Pipeline().setStages(Array(labelInt, features, decisionTree))

    val Array(trainDatas, testDatas) = orders_goods_int.randomSplit(Array(0.8, 0.2))

    val model = pipeline.fit(trainDatas)

    val testDF: DataFrame = model.transform(testDatas).persist(StorageLevel.MEMORY_AND_DISK)
    //    testDF.show()
    //    +---------+-----------+-----+------+-----+-----------+-------------+--------------------+----------+
    //    | memberId|productType|color|gender|label|   features|rawPrediction|         probability|prediction|
    //    +---------+-----------+-----+------+-----+-----------+-------------+--------------------+----------+
    //    | 13823611|         32|    2|     0|  0.0| [32.0,2.0]|  [269.0,0.0]|           [1.0,0.0]|       0.0|
    //    |  4034493|         12|    9|     0|  0.0| [12.0,9.0]| [1158.0,0.0]|           [1.0,0.0]|       0.0|
    //    |       62|         15|    9|     0|  0.0| [15.0,9.0]| [1158.0,0.0]|           [1.0,0.0]|       0.0|
    //    |       91|          0|   15|     1|  1.0| [0.0,15.0]|  [0.0,531.0]|           [0.0,1.0]|       1.0|
    //    |       91|         10|    5|     0|  0.0| [10.0,5.0]| [1158.0,0.0]|           [1.0,0.0]|       0.0|

    val trainDF = model.transform(trainDatas).persist(StorageLevel.MEMORY_AND_DISK)

    val allDF = trainDF.union(testDF)
      .select(
        'memberId,
        'prediction,
        when('prediction === 0, 1).otherwise(0) as "man",
        when('prediction === 1, 1).otherwise(0) as "woman"
      ).groupBy('memberId)
      .agg(sum("man").as("man"), sum("woman").as("woman"), count("memberId").as("all"))

    val leve5map: Map[String, Long] = leve5.collect().map(row=>{(row(0).toString,row(1).toString.toLong)}).toMap

    var getSex=udf((man:Double, woman:Double, all:Double)=>{
      var manp=man/all
      var womanp=woman/all
      var sz=0L
      if (manp>=0.6){
        sz=leve5map("0")
      }else if(womanp>=0.6){
        sz=leve5map("1")
      }else{
        sz=leve5map("-1")
      }
      sz
    })
    val jg = allDF.select('memberId.as("userId"), getSex('man cast(DoubleType), 'woman cast(DoubleType), 'all cast(DoubleType)).as("tagsId"))
    //    val evaluator = new MulticlassClassificationEvaluator()
    //      .setLabelCol("label")
    //      .setPredictionCol("prediction")
    //    val d = evaluator.evaluate(testDF)
    //
    //    println(">>>>>>"+d)
    //
    //    val xxx = model.stages(2).asInstanceOf[DecisionTreeClassificationModel]
    //    println(xxx.toDebugString)
    jg.show()
    null
  }

  def main(args: Array[String]): Unit = {
    exec()
  }
}
